Skip to content

[kafka] Handle basic Produce requests independently of storage - #4284

Draft
gyang94 wants to merge 9 commits into
apache:mainfrom
gyang94:codex/kafka-produce-protocol
Draft

gyang94 wants to merge 9 commits into
apache:mainfrom
gyang94:codex/kafka-produce-protocol

Conversation

@gyang94

@gyang94 gyang94 commented Sep 10, 2026

Copy link
Copy Markdown
Contributor

Purpose

Closes #4281. Part of #4185.

Implement non-idempotent Produce v3-v11 with a narrow backend contract. Validate acknowledgements, topic/partition identities and record batches; isolate invalid partitions, preserve response order and copy record data independently of the network buffer.

This is PR05, depending on #4261 (PR04 Metadata).

Brief change log

  • Valid requests preserve records, headers, listener context, acks and timeout.
  • Invalid batches and topics do not suppress valid partitions in the same request.
  • Transactional/idempotent/control batches and duplicate partition entries are rejected.
  • acks=0 emits no response and releases the request buffer on success and failure.
  • Production ApiVersions continues to expose only the existing capabilities until append is integrated.

Incremental scope: 5 files, +987 / -0 (621 implementation and 366 test lines). The implementation contract and its failure-path tests are kept together as one review unit; the line count includes Javadoc and license headers.

Tests

Java 11 at 936fd3f39e9e6fe595ed6bf8e30f0e2b5185ffe3:

  • mvn -o -pl fluss-kafka spotless:apply clean verify: 69 unit tests and 2 integration tests passed.
  • Checkstyle, Spotless, RAT and git diff --check passed.
  • The two existing integration tests cover Kafka transport and native DDL mapping; this stage does not activate production Produce.

The full repository test suite and Flink SQL runtime were not run.

API and Format

Adds independently testable non-idempotent Produce v3-v11 handling. Production API registration, native conversion and append follow in the later PRs.

Stack and review

Draft while prerequisites are unmerged. The PR targets Apache main, so the full Files changed view includes prerequisites. Use this incremental comparison to review this stage alone.

Generative AI disclosure: Codex assisted with extraction, implementation, tests and commit organization. Human review is required before merge.

gyang94 and others added 9 commits September 16, 2026 16:41
Introduce API registration, request context, version validation, and
asynchronous error mapping. Fix request buffer ownership and response
serialization cleanup while preserving the existing ApiVersions entry point.

Validated with mvn -o -pl fluss-rpc,fluss-kafka verify.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 555/555
AI-Contributed/UT: 525/525
Catch failures while mapping asynchronous handler results so the dispatcher
future completes exceptionally instead of remaining pending. Cover handler
futures that fail before and after dispatch registers its callback.

Validated with Java 11: Maven reactor build and 17 targeted Kafka tests,
including Checkstyle, Spotless, and license checks.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 27/27
AI-Contributed/UT: 28/28
Route requests through the dispatcher and advertise only implemented APIs.
Return version-aware errors for unsupported APIs and invalid requests.

Validated with mvn -o -pl fluss-kafka verify (23 unit tests and 1 IT).

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 313/313
AI-Contributed/UT: 137/137
Extract topic identity and the raw/string table mapping contract before Metadata. Validate table kinds, field projections and metadata columns independently of request handling and record decoding.

Validation: Java 11, mvn -o -pl fluss-kafka clean verify (47 unit tests and 2 integration tests); Checkstyle, Spotless and RAT passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 769/769
AI-Contributed/UT: 480/480
Resolve database.table topic names directly to Fluss table paths and return
qualified names for metadata. Validate name boundaries and cover same-named
tables in different databases through native metadata round trips.

Read projected field names directly from DataField to avoid rebuilding the
complete field-name list for every projected column.

Validated on Java 11: 67 Kafka unit tests and 2 integration tests passed,
along with Checkstyle, Spotless, RAT, and git diff --check.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 52/52
AI-Contributed/UT: 167/167
Integrate the DDL mapping prerequisite and use its resolver for Metadata discovery. Omit unsupported tables from all-topic queries and return per-topic mapping errors for named queries. Preserve metadata for compatible tables in mixed requests.

Validation: Java 11, mvn -o -pl fluss-kafka clean verify (62 unit tests and 2 integration tests); Checkstyle, Spotless and RAT passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 742/742
AI-Contributed/UT: 571/571
Route Metadata queries through the database.table mapping from PR03 and
enumerate compatible topics across databases. Preserve brokers and valid
topics when a requested database is absent or disappears during discovery;
propagate other gateway failures.

Cover qualified names, same-named tables across databases, synchronous and
asynchronous lookup failures, and Metadata v0-v11 over a real listener.

Validated on Java 11: 90 unit tests and 4 integration tests passed, along
with Checkstyle, Spotless, RAT, and git diff --check. Five additional
interleaved asynchronous review probes passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 129/129
AI-Contributed/UT: 514/514
Validate non-idempotent Produce v3-v11 and isolate invalid partitions before invoking a protocol-independent backend. Preserve partition order, acknowledgements and owned record data. Leave production registration to the append integration PR.

Validation: Java 11, mvn -o -pl fluss-kafka spotless:apply clean verify; 69 unit tests and 2 integration tests passed, including Checkstyle, Spotless and RAT.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 621/621
AI-Contributed/UT: 366/366
Read compressed records through a closeable streaming iterator to avoid
preallocating a record list from the declared batch count. Reject incoming
Kafka batches whose offset range disagrees with their record count before
copying records or invoking the backend. Keep Fluss-assigned offsets independent
of this input validation and preserve partition error isolation.

Cover all five codecs, owned record copies, malformed counts and offset ranges,
and backend-assigned offsets. Java 11 validation passed: 117 unit tests and
4 integration tests, plus Checkstyle, Spotless, RAT, and git diff --check.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 42/42
AI-Contributed/UT: 168/168
@gyang94
gyang94 force-pushed the codex/kafka-produce-protocol branch from 936fd3f to ff27c55 Compare September 17, 2026 09:04
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[kafka] Handle basic Produce requests independently of storage

1 participant